package com.atguigu.tabletest


import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.scala._
import org.apache.flink.table.api.{DataTypes, EnvironmentSettings, Table, TableEnvironment}
import org.apache.flink.table.descriptors.{Csv, FileSystem, Schema}


/**
 *
 * @description: 标准的tableApi操作
 * @time: 2021-03-15 15:35
 * @author: baojinlong
 **/
object Example02 {
  def main(args: Array[String]): Unit = {
    val environment: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // 设置并行度
    environment.setParallelism(1)

    // 创建表的执行环境
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(environment)
    // 1.1基于老版本Planner的流处理
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance
      .useOldPlanner
      .inStreamingMode
      .build
    val oldStreamTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(environment, settings)

    // 1.2基于老版本的批处理
    val batchEnv: ExecutionEnvironment = ExecutionEnvironment.getExecutionEnvironment
    BatchTableEnvironment.create(batchEnv)

    // 1.3基于 blink  planner的流处理
    val blinkStreamSettings: EnvironmentSettings = EnvironmentSettings.newInstance
      .useBlinkPlanner()
      .inStreamingMode
      .build
    val blinkStreamTableEnv: StreamTableEnvironment = StreamTableEnvironment.create(environment, blinkStreamSettings)

    // 1.4基于 blink  planner的批处理
    val blinkBatchSettings: EnvironmentSettings = EnvironmentSettings.newInstance
      .useBlinkPlanner()
      .inBatchMode
      .build
    val blinkBatchTableEnv: TableEnvironment = TableEnvironment.create(blinkBatchSettings)


    // 连接外部系统,读取文件注册表
    val filePath: String = "E:/big-data/FlinkTutorial/src/main/resources/sensor.data"
    tableEnv
      .connect(new FileSystem().path(filePath))
      .withFormat(new Csv())
      .withSchema(new Schema().field("id", DataTypes.STRING).field("timestamp", DataTypes.BIGINT).field("temperature", DataTypes.DOUBLE))
      .createTemporaryTable("inputTable")


    val inputTable: Table = tableEnv.from("inputTable")
    inputTable.toAppendStream[(String, Long, Double)].print("inputTableData")

    // 使用table api
    val sensorTable: Table = tableEnv.from("inputTable")
    val resultTable: Table = sensorTable
      .select('id, 'temperature)
      .filter('id === "sensor_1")

    // sql
    val resultSqlTable: Table = tableEnv.sqlQuery(
      """
        |select id,temperatue
        |from inputTable
        |where id = 'sensor_1'
        |""".stripMargin
    )

    val inputTable2: Table = tableEnv.from("kafkaInputTable")
    inputTable2.toAppendStream[(String, Long, Long)].print("kafkaInputTable")


    resultTable.toAppendStream[(String, Double)].print("result")
    resultSqlTable.toAppendStream[(String, Double)].print("sqlResult")

    environment.execute("table sql api")
  }

}
